/*
    Copyright 2010 Northbranchlogic, Inc.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.runjrun;

import java.util.*;
import com.norbl.util.*;
import com.norbl.util.ec2.*;
import com.norbl.util.s3.*;
import com.norbl.util.sqs.*;
import com.norbl.util.sdb.*;
import com.xerox.amazonws.sdb.SDBException;


/** <tt>CurrentStates</tt> runs the <tt>update</tt>() methods on the node and task states
 *  and maintains a snapshot of the current state data, including summary
 *  statistics. Optionally it requeues orphaned tasks.<p>
 *
 *  All the action is performed in the {@link #update()} method.
 *  This method is usually called in a loop in a separate thread, but it
 *  may also be called after an action has been performed (usually via a gui).<p>
 *
 *  <tt>CurrentStates</tt> also contains methods that perform operations
 *  such as clearing all tasks.  There methods are here because some
 *  are used by both {@link Client} and by the monitor gui.
 *
 * @author Barnet Wagman
 */
public class CurrentStates {

    private Ec2Ops eops;
    private S3Ops sops;
    private SdbOps dops;
    private SqsOps qops;

    boolean requeueOrphans;

    public NodeStateTable nodes;
    public TaskStateTable tasks;
    
    public NodeSummary nodeSummary;
    public TaskSummary taskSummary;

    List<StateChangeListener> listeners;

    public CurrentStates(Ec2Ops eops,S3Ops sops, SdbOps dops,SqsOps qops,
                        boolean requeueOrphans,
                        long runnerNodePulseIntervalMillis)
        throws SDBException {

        this.eops = eops;
        this.sops = sops;
        this.dops = dops;
        this.qops = qops;        
        this.requeueOrphans = requeueOrphans;
     
        nodes = new NodeStateTable(dops, eops, runnerNodePulseIntervalMillis);
        tasks = new TaskStateTable(dops);

        nodeSummary = new NodeSummary();
        taskSummary = new TaskSummary(requeueOrphans);

        listeners = new ArrayList<StateChangeListener>();
    }

       // -----------------------------------

    synchronized public void update() {       
        updateUnsynchronized();
    }

    private void updateUnsynchronized() {
        try {
            boolean nodesModified = true;
            List<TaskState> orphans;
            do {             
                nodesModified = nodes.update();                
                orphans = tasks.update(nodes.currentStates,sops, qops);                
                if ( orphans.size() > 0 ) {
                    if ( requeueOrphans ) {
                        tasks.requeueOrphans(orphans);
                        taskSummary.nRequeued += orphans.size();
                    }
                    else {
                        tasks.recordOrphanedTasks(orphans);
                    }
                }              
            } while( nodesModified || (orphans.size() > 0));

            nodeSummary.summarize(nodes.currentStates);            
            taskSummary.summarize(tasks.currentStates);           
            notifyListeners();           
        }
        catch(Throwable x) {
            System.out.println("CurrentStates.update():\n" +
                    StringUtil.throwableToString(x) +
                    "^^^^^^^^^^^^^^^^");
        }
    }

    public void addStateChangeListner(StateChangeListener l) {
        listeners.add(l);
    }

    public void notifyListeners() {       
        for ( StateChangeListener l : listeners ) {
            l.stateChanged();            
        }
    }

       // ------------------------------------

    public class NodeSummary {
        public int nNodes;
        public int nNodesIdle;
        public int nNodesActive;
        public int nNodesNil;
        public int nNodesNoPulse;

        void summarize(NodeState[] els) {

            nNodes = els.length;
            nNodesIdle = 0;
            nNodesActive = 0;
            nNodesNil = 0;
            nNodesNoPulse = 0;

            for (NodeState el : els ) {
                NodeState.Status s = el.status;
                if ( s.equals(NodeState.Status.active) )  ++nNodesActive;
                if ( s.equals(NodeState.Status.idle) ) ++nNodesIdle;
                if ( s.equals(NodeState.Status.nil) ) ++nNodesNil;
                if ( s.equals(NodeState.Status.no_pulse) ) ++nNodesNoPulse;
            }
        }

        public String toString() {
            return( "n nodes=" + nodes + " idle=" + nNodesIdle +
                    " active=" + nNodesActive + " nil=" + nNodesNil +
                    " no pulse=" + nNodesNoPulse);
        }
    }

    public class TaskSummary {
        public int nTasks;
        public int nTasksWaiting;
        public int nTasksRunning;
        public int nTasksFinished;
        public int nTasksTerminatedByException;
        public int nTasksOrphaned;
        public int nTasksMissingRunnable;
        public int nRequeued;
        public boolean autoRequeueOrphans;

        TaskSummary(boolean autoRequeue) {
            nRequeued = 0;
            TaskSummary.this.autoRequeueOrphans = autoRequeue;
        }

        void summarize(TaskState[] els) {

            nTasks = els.length;
            nTasksWaiting = 0;
            nTasksRunning = 0;
            nTasksFinished = 0;
            nTasksTerminatedByException = 0;
            nTasksOrphaned = 0;
            nTasksMissingRunnable = 0;

            for ( TaskState el : els ) {
                TaskState.Status s = el.status;
                if ( s.equals(TaskState.Status.waiting) ) ++nTasksWaiting;
                if ( s.equals(TaskState.Status.running) ) ++nTasksRunning;
                if ( s.equals(TaskState.Status.finished) ) ++nTasksFinished;
                if ( s.equals(TaskState.Status.terminatedByException) )
                    ++nTasksTerminatedByException;
                if ( s.equals(TaskState.Status.orphan) )
                    ++nTasksOrphaned;
                if ( s.equals(TaskState.Status.missingRunnable) )
                    ++nTasksMissingRunnable;
            }
        }

        public String toString() {
            return("n tasks=" + nTasks + " waiting=" + nTasksWaiting +
                    " running=" + nTasksRunning + " finished=" + nTasksFinished +
                    " termEx=" + nTasksTerminatedByException +
                    " orphans=" + nTasksOrphaned +
                    " missing runnables=" + nTasksMissingRunnable +
                    " requeued=" + nRequeued);
        }
    }

        /** For testing only.
         *
         * @return string containing n nodes, etc.
         */
    public String toString() {
        return( "N nodes=" + nodes.currentStates.length + " " +
                " tasks=" + tasks.currentStates.length + "\n" +
                "- nodes: " + nodeSummary.toString() + "\n" +
                "- tasks: " + taskSummary.toString() );
    }  
}